package tbdp.tool;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.log4j.Logger;
import tbdp.tool.common.Constants;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;

/**
 * Created by xbkaishui on 16/10/15.
 * 把数据上传到hdfs
 */
public class Uploader {

    private static Logger logger = Logger.getLogger(Uploader.class);

    private static Configuration conf = new Configuration();

    public static void upload(String input, String output) {
        upload(input, output, Constants.defaultCodec);
    }


    public static void upload(String input, String output, String coderClass) {
        try {
            Class<?> codecClass = Class.forName(coderClass);
            FileSystem fs = FileSystem.get(conf);
            CompressionCodec codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
//            File localFile = new File(input);
//            String fileName = localFile.getName();
//            String tmpOutput = "hdfs:/tmp/" + marketType + "/" + fileName;
//            logger.info(String.format("src file %s  tmp %s ", input, tmpOutput));
            //指定压缩文件路径
//            fs.copyFromLocalFile(false, new Path(input), new Path(tmpOutput));

            FSDataOutputStream outputStream = fs.create(new Path(output));
            InputStream in = new FileInputStream(input);
//            FSDataInputStream in = fs.open(new Path(tmpOutput));
            //创建压缩输出流
            CompressionOutputStream out = codec.createOutputStream(outputStream);
            IOUtils.copyBytes(in, out, conf);
            IOUtils.closeStream(in);
            IOUtils.closeStream(out);
        } catch (Exception e) {
            throw new RuntimeException("upload file error", e);
        }

    }


    public static void main(String[] args) {
//        System.out.println("haha");
        String input = "/tmp/txn_cl_train_model_01.xml";
        String output = "hdfs:/tmp/txn_cl_train_model_0111.gz";
//        upload(input, output, Constants.sh, Constants.defaultCodec);

        upload(input, output, Constants.defaultCodec);

    }


}
